package com.citusdata.elven.router;

import com.citusdata.elven.router.Endpoint;
import com.citusdata.elven.router.EndpointHealthMonitor;
import com.citusdata.elven.util.ConfigManager;
import java.lang.IllegalArgumentException;
import java.lang.IllegalStateException;
import java.text.SimpleDateFormat;
import java.text.ParseException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.List;
import java.util.Set;
import java.util.NavigableMap;
import java.util.TreeMap;
import org.apache.commons.configuration.Configuration;
import org.apache.log4j.Logger;


// Routing component that maps any given <partition key, timestamp> pair to a
// set of endpoints. Internally, this class parses a list of membership lists
// from a configuration file. Each parsed membership list contains two sets of
// information: (1) an activation timestamp that notes the point in time this
// list becomes active, and (2) the list of endpoints that will be active for
// the specified time period.
// This particular implementation uses replication groups to map partition key
// and timestamp pairs. That is, data is partitioned and routed to a particular
// group for a given time interval, and this group contains a fixed set of
// replicas. This policy works well for Stream Processing Engines (SPEs).

public class ReplicationGroupRouter {
  private static final Logger logger = Logger.getLogger(ReplicationGroupRouter.class);

  protected static final String SEPARATOR = ":";
  protected static final long   EPOCH_START = 0L;  // milliseconds

  // Optional monitoring component that tracks endpoint availability. This
  // component also acts as a post-interceptor for getEndpoints(), and
  // temporarily filters out endpoints that had failures. 
  protected final EndpointHealthMonitor healthMonitor;

  protected NavigableMap timesSnapshots = new TreeMap();
  protected NavigableMap timesReplicationFactors = new TreeMap();


  public ReplicationGroupRouter(String configName) {
    Configuration config = ConfigManager.getPropertiesConfig(configName);
    if (config == null) {
      // If config load failed, the related exception has already been logged. 
      throw new IllegalStateException("Could not load essential configuration");
    }

    // Check from configuration if health monitoring and throttling are enabled.
    boolean enabled = config.getBoolean("Router.EndpointHealthMonitor.Enabled", false);
    if (enabled) {
      healthMonitor = new EndpointHealthMonitor();
    } else {
      healthMonitor = null;
    }

    // Read and initialize membership information.
    List snapshotList = config.getList("Router.SnapshotList");
    for (int sp = 0; sp < snapshotList.size(); ++sp) {
      String snapshotName = (String) snapshotList.get(sp);
      String snapshotKey  = "Router." + snapshotName + ".";
      
      String activeFromString = config.getString(snapshotKey + "ActiveFrom");
      long   activeFrom = 0;

      try {
        // Year-month-day hour:minute:second timezone format
        SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss zzz");
        Date date = format.parse(activeFromString);
        activeFrom = date.getTime();

      } catch(ParseException ex) {
        // Mis-configured dates can potentially corrupt all endpoints across all
        // time intervals; fail hard and early.
        logger.fatal("Exception when parsing snapshot: " + snapshotName, ex);
        throw new IllegalStateException("Illegal routing configuration, failing");
      }

      boolean zoneAware = config.getBoolean(snapshotKey + "ZoneAware", false);
      int replicationFactor = config.getInt(snapshotKey + "ReplicationFactor", 1);

      List snapshot = new ArrayList();
      List endpointNames = config.getList(snapshotKey + "Endpoint");

      for (int ep = 0; ep < endpointNames.size(); ++ep) {
        String endpointName = (String) endpointNames.get(ep);
        Endpoint endpoint = parseEndpoint(endpointName, zoneAware);
        if (endpoint != null) {
          snapshot.add(endpoint);
        } else {
          logger.warn("Failed to parse endpoint name: " + endpointName);
        }
      }  // for endpoint index

      if (snapshot.size() >= replicationFactor) {
        Collections.sort(snapshot);
        timesSnapshots.put(activeFrom, snapshot);
        timesReplicationFactors.put(activeFrom, replicationFactor);
      } else {
        // If snapshot configuration does not contain enough endpoints, the
        // router will not be able to route accurately; fail hard and early.
        logger.fatal("Not enough endpoints in snapshot: " + snapshotKey);
        throw new IllegalStateException("Insufficient number of endpoints, failing");
      }

    }  // for snapshot index
  }


  private static Endpoint parseEndpoint(String line, boolean zoneAware) {
    Endpoint endpoint = null;
    
    // Parse the configuration line, and if the configuration is valid, create
    // an endpoint.
    int first  = line.indexOf(SEPARATOR);
    int second = line.lastIndexOf(SEPARATOR);

    if (zoneAware) {
      if ( (0 <= first) && (first < second) && (second+1 < line.length()) ) {
        String hostname   = line.substring(0, first);
        String portString = line.substring(first+1, second);
        String zone = line.substring(second+1);

        int port = Integer.parseInt(portString);
        endpoint = new Endpoint(hostname, port, zone);
      } 
    } else {
      if ( (0 <= first) && (first+1 < line.length()) ) { 
        String hostname   = line.substring(0, first);
        String portString = line.substring(first+1);

        int port = Integer.parseInt(portString);
        endpoint = new Endpoint(hostname, port);
      }
    }

    return endpoint;
  }

  
  public List getActivationTimes(long startTime, long endTime) {
    // Hack: Handle timestamps before Epoch differently for Linear Road's
    // expenditure requests.
    if ( (startTime < 0) && (endTime < 0) ) {
      Long firstKey = (Long) timesSnapshots.firstKey();
      List negativeTimes = new ArrayList();
      negativeTimes.add(firstKey);

      return negativeTimes;
    }

    // getActivationTimes() includes activation times for the union of two sets
    // of snapshots: (1) the last snapshot to become active before startTime,
    // including startTime, and (2) all snapshots that are active between
    // startTime and endTime, excluding startTime. 
    // For example, assume two snapshots with activation times 120 and 250. In
    // this setting, the first snapshot represents the interval [120, 250[ and
    // the second one represents [250, inf[. As a result, a range query for
    // 210-250 will include snapshots for times 120 and 250.
    List activationTimes = new ArrayList();
    Object firstTime = timesSnapshots.floorKey(startTime);
    if (firstTime != null) {
      activationTimes.add(firstTime);
    }

    boolean inclusive = true;
    NavigableMap intervalMap = timesSnapshots.subMap(startTime, !(inclusive), 
                                                     endTime, inclusive);
    Set intervalSet = intervalMap.keySet();
    activationTimes.addAll(intervalSet);

    return activationTimes;
  }


  public List getEndpoints(String partitionKey, long activationTime) {
    // hashCode() performs poorly if the key contains more than 16 characters.
    // This class provides another interface in case the client chooses to use
    // a different hash function.
    int hashCode = partitionKey.hashCode();
    List endpoints = getEndpoints(hashCode, activationTime);

    return endpoints;
  }


  public List getEndpoints(int hashCode, long activationTime) {
    if (!timesSnapshots.containsKey(activationTime)) {
      throw new IllegalArgumentException("Non-existing activationTime: " + 
                                         activationTime);
    }

    List snapshot = (List) timesSnapshots.get(activationTime);
    Integer replicationFactor = (Integer) timesReplicationFactors.get(activationTime);

    // If zone awareness is enabled, endpoints in snapshot lists are sorted by
    // zone. To determine the endpoints in a replication group, we use the
    // following algorithm:
    // (1) Assume snapshot list contains n elements. We "divide" the list into r
    // segments, where r is the replication factor.
    // (2) Each segment contains (n/r) elements. We map the partition key (hash
    // code) to an index between 0 and (n/r)-1.
    // (3) We pick the indexed element (endpoint) from each segment, and return.

    int segmentSize = snapshot.size() / replicationFactor;
    int segmentIndex = Math.abs(hashCode % segmentSize);

    List endpoints = new ArrayList();
    for (int i = 0; i < replicationFactor; ++i) {
      int replicaIndex = segmentIndex + (i*segmentSize);
      Endpoint replica = (Endpoint) snapshot.get(replicaIndex);
      endpoints.add(replica);
    }

    // Post-process endpoints, and remove any that are currently unavailable.
    List healthyEndpoints = healthyEndpoints(endpoints);
    return healthyEndpoints;
  }


  public List healthyEndpoints(List endpoints) {
    if (healthMonitor == null) {
      return endpoints;
    }

    long currentTime = System.currentTimeMillis();
    List healthyEndpoints = new ArrayList();

    for (int i = 0; i < endpoints.size(); ++i) {
      Endpoint endpoint = (Endpoint) endpoints.get(i);
      if (!healthMonitor.throttle(endpoint, currentTime)) {
        healthyEndpoints.add(endpoint);
      }
    }
    
    return healthyEndpoints;
  }


  public void recordFailure(Endpoint endpoint) {
    if (healthMonitor == null) {
      return;  // No interceptors to process this failure information
    }

    long failureTime = System.currentTimeMillis();
    healthMonitor.recordFailure(endpoint, failureTime);
  }
}
